package com.demo.task.socket

import android.util.Log
import com.demo.task.MainApplication
import com.demo.task.ext.getSerial
import com.demo.task.ext.getVersion
import com.demo.task.ext.toJson
import com.google.gson.Gson
import com.orhanobut.logger.Logger
import info.mqtt.android.service.MqttAndroidClient
import info.mqtt.android.service.QoS
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.runBlocking
import org.eclipse.paho.client.mqttv3.IMqttActionListener
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken
import org.eclipse.paho.client.mqttv3.IMqttToken
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended
import org.eclipse.paho.client.mqttv3.MqttConnectOptions
import org.eclipse.paho.client.mqttv3.MqttMessage

class MqttManager(
    private val builder: Builder
) : RemoteInterface {

    private val mqttClient by lazy {
        MqttAndroidClient(
            MainApplication.instance,
            builder.url.orEmpty(),
            getSerial() + "_mqtt"
        )
    }

    private val mqttOption by lazy {
        MqttConnectOptions().apply {
            isCleanSession = false
            isAutomaticReconnect = true
            maxReconnectDelay = 5_000
            connectionTimeout = 10
            keepAliveInterval = 5

            setWill(
                "topic/${getSerial()}",
                mapOf("msg" to "offline").toJson().toByteArray(),
                QoS.AtLeastOnce.value,
                true
            )
        }
    }


    private val _msg = MutableSharedFlow<String>()
    override val msg: SharedFlow<String> get() = _msg

    private val _isConnect = MutableStateFlow(false)
    override val isConnect: StateFlow<Boolean> get() = _isConnect

    override fun connect() {
        mqttClient.addCallback(object : MqttCallbackExtended {
            override fun connectionLost(cause: Throwable?) {
                Log.e(TAG, "connectionLost: ", cause)
                runBlocking { _isConnect.emit(false) }

            }

            override fun messageArrived(topic: String?, message: MqttMessage?) {
            }

            override fun deliveryComplete(token: IMqttDeliveryToken?) {
            }

            override fun connectComplete(reconnect: Boolean, serverURI: String?) {
                Log.e(TAG, "connectComplete: $reconnect $serverURI")

                runBlocking { _isConnect.emit(true) }
                subscribeTopic()
                sendHeartbeat()
                sendVersion()
            }

        })
        mqttClient.connect(mqttOption, null, object : IMqttActionListener {
            override fun onSuccess(asyncActionToken: IMqttToken?) {
                Log.w(TAG, "onSuccess: ${asyncActionToken?.client?.clientId}")
                //runBlocking { _isConnect.emit(true) }

            }

            override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) {
                //runBlocking { _isConnect.emit(false) }
                Log.e(TAG, "onFailure: ${asyncActionToken?.client?.clientId}", exception)

            }

        })
    }

    private fun subscribeTopic() {
        mqttClient.subscribe(
            "device/${getSerial()}",
            QoS.AtLeastOnce.value
        ) { _, msg ->
            Log.w(TAG, "subscribeTopic: $msg")
            Logger.w("MqttManager 接收消息: $msg")
            runBlocking { _msg.emit(msg.toString()) }
        }

    }

    override fun send(msg: String) {
        publishMsg(msg)
    }

    override fun sendBasic(map: Map<String, Any>) {
        val message = mapOf(
            "type" to "basic",
            "message" to map
        )
        publishMsg(Gson().toJson(message))
    }

    private fun sendVersion() {
        val message = mapOf(
            "type" to "basic",
            "message" to mapOf("androidv" to getVersion())
        )
        publishMsg(Gson().toJson(message))
    }

    private fun sendHeartbeat() {
        val message = mapOf(
            "type" to "ping",
            "message" to "online"
        )
        publishMsg(Gson().toJson(message))
    }

    private fun publishMsg(msg: String) {
        Log.w(TAG, "publishMsg: $msg")
        Logger.w("MqttManager 发送消息: $msg")
        val mqttMessage = MqttMessage().apply {
            qos = QoS.AtLeastOnce.value
            payload = msg.toJson().toByteArray()
        }

        if (mqttClient.isConnected) {
            mqttClient.publish(
                "topic/${getSerial()}",
                mqttMessage
            )
        }
    }

    override fun close() {
        mqttClient.disconnect()
    }

    class Builder {
        internal var url: String? = null

        fun url(url: String) = apply {
            this.url = url
        }

        fun build(): MqttManager {
            requireNotNull(this.url) { "URL must be set" }
            return MqttManager(this)
        }
    }

    companion object {
        private const val TAG = "MqttManager"
    }
}